package geotrellis.chatta

import com.typesafe.config.ConfigFactory
import geotrellis.spark.util.SparkUtils
import org.apache.spark.{SparkConf, SparkContext}

object ChattaPipelineJson {
  import geotrellis.spark._
  import geotrellis.spark.pipeline._
  import geotrellis.spark.pipeline.ast._

  implicit val sc = SparkUtils.createSparkContext(
    "GeoTrellis ETL SinglebandIngest",
    new SparkConf(true)
      .setMaster(ConfigFactory
        .load()
        .getString("spark.master")))
  // pipeline json example
  val maskJson =
    """
      |[
      |  {
      |    "uri" : "data/arg_wm/",
      |    "type" : "singleband.spatial.read.hadoop"
      |  },
      |  {
      |    "resample_method" : "nearest-neighbor",
      |    "type" : "singleband.spatial.transform.tile-to-layout"
      |  },
      |  {
      |    "crs" : "EPSG:3857",
      |    "scheme" : {
      |      "crs" : "epsg:3857",
      |      "tileSize" : 256,
      |      "resolutionThreshold" : 0.1
      |    },
      |    "resample_method" : "nearest-neighbor",
      |    "type" : "singleband.spatial.transform.buffered-reproject"
      |  },
      |  {
      |    "end_zoom" : 0,
      |    "resample_method" : "nearest-neighbor",
      |    "type" : "singleband.spatial.transform.pyramid"
      |  },
      |  {
      |    "name" : "mask",
      |    "uri" : "file:/E:/geotrellis-chatta-demo/service/geotrellis/data/chatta-demo-pipeline",
      |    "key_index_method" : {
      |      "type" : "zorder"
      |    },
      |    "scheme" : {
      |      "crs" : "epsg:3857",
      |      "tileSize" : 256,
      |      "resolutionThreshold" : 0.1
      |    },
      |    "type" : "singleband.spatial.write"
      |  }
      |]
    """.stripMargin

  def main(args: Array[String]): Unit = {
    // parse the JSON above
    val list: Option[Node[Stream[(Int, TileLayerRDD[SpatialKey])]]] = maskJson.node

    list match {
      case None => println("Couldn't parse the JSON")
      case Some(node) => {
        // eval evaluates the pipeline
        // the result type of evaluation in this case would ben Stream[(Int, TileLayerRDD[SpatialKey])]
        node.eval.foreach { case (zoom, rdd) =>
          println(s"ZOOM: ${zoom}")
          println(s"COUNT: ${rdd.count}")
        }
      }
    }
  }

}
